package com.example.utils;
import org.springframework.beans.factory.DisposableBean;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
/*
@description:延时队列工具类
@ClassName DelayTaskProcessorUtils
@author chen
@create 2025-08-14 16:10
@Version 1.0
*/
public class DelayTaskProcessorUtils implements Runnable, DisposableBean
{

    // 名称，标识作用
    private String name;
    // 容量
    private int capacity;

    private int concurrency;

    //延迟队列
    private DelayQueue<DelayTask> delayQueue = new DelayQueue<>();

    //是否已停止
    private volatile boolean stop = false;

    // 当前待处理的任务数量
    private AtomicInteger size = new AtomicInteger(0);

    private List<Thread> consumerThreadList;



    /*
     * @param name        名称
     * @param capacity    队列容量
     * @param concurrency 并发数
     * @==============History===============<br/>;
     * @Description //   延时任务执行器
     * @Date 2025/8/14
     * @Author < a href="chencheng@prmt.net">chen</ a>;
     */
    public DelayTaskProcessorUtils(String name, int capacity, int concurrency)
    {
        if (capacity <= 0)
        {
            throw new IllegalArgumentException("Illegal Capacity: " + capacity);
        }
        if (concurrency <= 0)
        {
            throw new IllegalArgumentException("Illegal Concurrency: " + concurrency);
        }
        this.name = name;
        this.capacity = capacity;
        this.concurrency = concurrency;

        // 创建 concurrency 个消费者线程，并启动，消费者线程会从延迟队列拉取到期的任务进行消费
        this.consumerThreadList = new ArrayList<>(this.concurrency);
        for (int i = 0; i < this.concurrency; i++)
        {
            String consumerThreadName = this.generateConsumerThreadName(i);
            Thread consumerThread = new Thread(this, consumerThreadName);
            consumerThread.start();
            this.consumerThreadList.add(consumerThread);
        }
    }

    protected String generateConsumerThreadName(int consumerIndex)
    {
        return String.format("DelayTaskProcessor-ConsumerThread-%s-%s", name, consumerIndex);
    }



    /*
     * @param delayTime          延迟时间
     * @param delayTimeUnit      延迟时间单位
     * @param task               任务
     * @==============History===============<br/>;
     * @Description //  放入要执行的延迟任务
     * @Date 2025/8/14
     * @Author < a href="chencheng@prmt.net">chen</ a>;
     */
    public boolean put(long delayTime, TimeUnit delayTimeUnit, Runnable task)
    {
        long taskExecuteTimeMs = System.currentTimeMillis() + delayTimeUnit.toMillis(delayTime);
        return this.put(taskExecuteTimeMs, task);
    }


    /*
     * @param taskExecuteTimeMs  任务什么时候执行
     * @param task               任务
     * @==============History===============<br/>;
     * @Description //
     * @Date 2025/8/14
     * @Author < a href="chencheng@prmt.net">chen</ a>;
     */
    public boolean put(long taskExecuteTimeMs, Runnable task)
    {
        if (size.get() == capacity)
        {
            return false;
        }
        synchronized (this)
        {
            //当前待处理的任务数量达到了容量，返回false
            if (size.get() == capacity)
            {
                return false;
            }
            //放入延迟队列
            this.delayQueue.put(new DelayTask(taskExecuteTimeMs, task));
            //待处理任务数+1
            size.incrementAndGet();
        }
        return true;
    }


    public void shutdown()
    {
        this.destroy();
    }


    @Override
    public void run()
    {
        while (true)
        {
            if (stop)
            {
                break;
            }
            Runnable task = null;
            try {
                // 从延迟队列中拿到任务
                task = delayQueue.take().getTask();
                //容量--
                size.decrementAndGet();
            }
            catch (InterruptedException e)
            {
                Thread.currentThread().interrupt();
            }
            if (task == null)
            {
                continue;
            }
            this.taskDispose(task);
        }
    }

    /*
     * @param task
     * @==============History===============<br/>;
     * @Description //   处理任务
     * @Date 2025/8/14
     * @Author < a href="chencheng@prmt.net">chen</ a>;
     */
    private void taskDispose(Runnable task)
    {
        Throwable throwable = null;
        try
        {
            task.run();
        }
        catch (Throwable t)
        {
            throwable = t;
        }
        finally
        {
            this.afterExecute(task, throwable);
        }
    }

    /*
     * @param
     * @==============History===============<br/>;
     * @Description //   任务执行完成后回调
     * @Date 2025/8/14
     * @Author < a href="chencheng@prmt.net">chen</ a>;
     */
    private void afterExecute(Runnable task, Throwable throwable)
    {

    }

    @Override
    public void destroy()
    {
        this.stop = true;
        for (Thread consumerThread : consumerThreadList)
        {
            consumerThread.interrupt();
        }
    }
}
